package com.dayang.redbook.user.relation.biz.consumer;

import cn.hutool.core.date.DateUtil;
import com.dayang.framework.common.util.JsonUtil;
import com.dayang.redbook.user.relation.biz.constant.MQConstants;
import com.dayang.redbook.user.relation.biz.constant.RedisKeyConstants;
import com.dayang.redbook.user.relation.biz.domain.dataobject.FansDO;
import com.dayang.redbook.user.relation.biz.domain.dataobject.FollowingDO;
import com.dayang.redbook.user.relation.biz.domain.mapper.FansDOMapper;
import com.dayang.redbook.user.relation.biz.domain.mapper.FollowingDOMapper;
import com.dayang.redbook.user.relation.biz.enums.FollowUnfollowTypeEnum;
import com.dayang.redbook.user.relation.biz.model.dto.CountFollowUnfollowMqDTO;
import com.dayang.redbook.user.relation.biz.model.dto.FollowUserMqDTO;
import com.dayang.redbook.user.relation.biz.model.dto.UnfollowUserMqDTO;
import com.dayang.redbook.user.relation.biz.rpc.DistributedIdGeneratorRpcService;
import com.google.common.util.concurrent.RateLimiter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Objects;

/**
 * @author: dayang
 * @since: 2024/12/18
 * @description: 关注、取关 MQ 消费者
 */
@Component
@RocketMQMessageListener(
        consumerGroup = "xiaohashu_group" + MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW,
        topic = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW,
        consumeMode = ConsumeMode.ORDERLY
)
@Slf4j
@RequiredArgsConstructor
public class FollowUnfollowConsumer implements RocketMQListener<Message> {
    private final FollowingDOMapper followingDOMapper;
    private final FansDOMapper fansDOMapper;
    private final TransactionTemplate transactionTemplate;
    private final RedisTemplate<String,Object> redisTemplate;
    private final RateLimiter rateLimiter;
    private final RocketMQTemplate rocketMQTemplate;
    private final DistributedIdGeneratorRpcService idGeneratorRpcService;
    @Override
    public void onMessage(Message message) {
        // 流量削峰：通过获取令牌，如果没有令牌可用，将阻塞，直到获得
        rateLimiter.acquire();
        // 消息体
        String bodyJsonStr = new String(message.getBody());
        // 标签
        String tags = message.getTags();
        log.info("==> FollowUnfollowConsumer 消费了消息 {}, tags: {}", bodyJsonStr, tags);

        // 根据 MQ 标签，判断操作类型
        if (Objects.equals(tags,MQConstants.TAG_FOLLOW)){
            handleFollowTagMessage(bodyJsonStr);
        }else if (Objects.equals(tags,MQConstants.TAG_UNFOLLOW)){
            handleUnfollowTagMessage(bodyJsonStr);
        }
    }

    /**
     * 取消关注
     * @param bodyJsonStr
     */
    private void handleUnfollowTagMessage(String bodyJsonStr) {
        // 将消息体 Json 字符串转为 DTO 对象
        UnfollowUserMqDTO unfollowUserMqDTO = JsonUtil.parseObject(bodyJsonStr, UnfollowUserMqDTO.class);
        // 判空
        if (Objects.isNull(unfollowUserMqDTO)) return;

        Long userId = unfollowUserMqDTO.getUserId();
        Long unfollowUserId = unfollowUserMqDTO.getUnfollowUserId();
        LocalDateTime updateTime = unfollowUserMqDTO.getUpdateTime();


        // 编程式提交事务
        boolean isSuccess = Boolean.TRUE.equals(transactionTemplate.execute(status -> {
            try {
                int count = followingDOMapper.inserOrUpdate(FollowingDO.builder()
                        .id(Long.valueOf(idGeneratorRpcService.getSnowflakeId()))
                        .userId(userId)
                        .followingUserId(unfollowUserId)
                        .updateTime(updateTime)
                        .type(0)
                        .build());
                // 粉丝表：一条记录
                if (count > 0) {
                    fansDOMapper.insertOrUpdate(FansDO.builder()
                            .userId(unfollowUserId)
                            .fansUserId(userId)
                            .type(0)
                            .updateTime(updateTime)
                            .build()
                    );
                }
                return true;
            } catch (Exception ex) {
                status.setRollbackOnly(); // 标记事务为回滚
                log.error("", ex);
            }
            return false;
        }));
        // 若数据库删除成功，更新 Redis，将自己从被取注用户的 ZSet 粉丝列表删除
        if (isSuccess) {
            // 被取关用户的粉丝列表 Redis Key
            String fansRedisKey = RedisKeyConstants.buildUserFansKey(unfollowUserId);
            // 删除指定粉丝
            redisTemplate.opsForZSet().remove(fansRedisKey, userId);

            // 发送 MQ 通知计数服务：统计关注数
            CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
                    .userId(userId)
                    .targetUserId(unfollowUserId)
                    .type(FollowUnfollowTypeEnum.UNFOLLOW.getCode()) // 取关
                    .build();
            sendMQ(countFollowUnfollowMqDTO);

        }
    }

    /**
     * 关注
     * @param bodyJsonStr
     */
    private void handleFollowTagMessage(String bodyJsonStr) {
        FollowUserMqDTO followUserMqDTO = JsonUtil.parseObject(bodyJsonStr, FollowUserMqDTO.class);
        if (Objects.isNull(followUserMqDTO)) return;
        // 幂等性通过联合唯一索引保证
        Long followUserId = followUserMqDTO.getFollowUserId();
        Long userId = followUserMqDTO.getUserId();
        LocalDateTime updateTime = followUserMqDTO.getUpdateTime();
        boolean isSuccess = Boolean.TRUE.equals(transactionTemplate.execute(status -> {
            try {
                // 分别向关注表和粉丝表各插入一条记录
                int count = followingDOMapper.inserOrUpdate(FollowingDO.builder()
                        .id(Long.valueOf(idGeneratorRpcService.getSnowflakeId()))
                        .userId(userId)
                        .followingUserId(followUserId)
                        .updateTime(updateTime)
                        .type(1)
                        .build());
                // 粉丝表：一条记录
                if (count > 0) {
                    fansDOMapper.insertOrUpdate(FansDO.builder()
                            .id(Long.valueOf(idGeneratorRpcService.getSnowflakeId()))
                            .userId(followUserId)
                            .fansUserId(userId)
                            .type(1)
                            .updateTime(updateTime)
                            .build()
                    );
                }
                return true;

            } catch (Exception ex) {
                status.setRollbackOnly(); // 标记事务为回滚
                log.error("", ex);
            }
            return false;
        }));
        if (isSuccess){
            // Lua 脚本
            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            script.setLocation(new ClassPathResource("lua/follow_check_and_update_fans_zset.lua"));
            script.setResultType(Long.class);
            // 时间戳
            long timestamp = DateUtil.toInstant(updateTime).toEpochMilli();
            // 构建被关注用户的粉丝列表 Redis Key
            String fansRedisKey = RedisKeyConstants.buildUserFansKey(followUserId);
            // 执行脚本
            redisTemplate.execute(script, Collections.singletonList(fansRedisKey), userId, timestamp);

            // 发送MQ通知计数服务:统计关注数
            CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
                    .userId(userId)
                    .targetUserId(followUserId)
                    .type(FollowUnfollowTypeEnum.FOLLOW.getCode()) // 关注
                    .build();
            sendMQ(countFollowUnfollowMqDTO);

        }

    }

    /**
     * 发送MQ通知计数服务
     * @param countFollowUnfollowMqDTO
     */
    private void sendMQ(CountFollowUnfollowMqDTO countFollowUnfollowMqDTO) {
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(JsonUtil.toJsonString(countFollowUnfollowMqDTO)).build();
        // 异步发送 MQ 消息
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FOLLOWING, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：关注数】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：关注数】MQ 发送异常: ", throwable);
            }
        });

        // 发送 MQ 通知计数服务：统计粉丝数
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FANS, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：粉丝数】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：粉丝数】MQ 发送异常: ", throwable);
            }
        });
    }
}
